package com.gitee.xhs.savePointReader.keyedState.valueStateReader;

import com.gitee.xhs.keyedState.valueState.ValueStateConstant;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.state.api.functions.KeyedStateReaderFunction;
import org.apache.flink.util.Collector;

/**
 * @author BigHead
 * @date 2022/6/29 14:50
 * @description ValueStateReaderFunction
 */
public class ValueStateReaderFunction extends KeyedStateReaderFunction<Long, ValueStateExhibitor> {

    private final String jobId;
    private final String operatorUid;
    private transient ValueState<Integer> valueState;

    public ValueStateReaderFunction(String jobId, String operatorUid) {
        this.jobId = jobId;
        this.operatorUid = operatorUid;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        valueState = getRuntimeContext().getState(ValueStateConstant.VALUE_STATE_DESCRIPTOR);
    }

    @Override
    public void readKey(Long key, Context ctx, Collector<ValueStateExhibitor> out) throws Exception {
        ValueStateExhibitor valueStateExhibitor = new ValueStateExhibitor();
        valueStateExhibitor.setJobId(jobId);
        valueStateExhibitor.setOperatorUid(operatorUid);
        valueStateExhibitor.setKey(key);
        valueStateExhibitor.setValue(valueState.value());
        out.collect(valueStateExhibitor);
    }
}
